package com.swak.paxos.impl;

import java.util.concurrent.ScheduledFuture;

import com.swak.paxos.PaxosConfig;
import com.swak.paxos.Proposer;
import com.swak.paxos.Value;
import com.swak.paxos.breakpoint.ProposerBP;
import com.swak.paxos.enums.ProposalType;
import com.swak.paxos.protocol.PaxosMessage;
import com.swak.paxos.state.BallotNumber;
import com.swak.paxos.state.BallotStatistics;
import com.swak.paxos.state.ProposerState;
import com.swak.paxos.state.TimeStat;
import com.swak.paxos.store.PaxosStorage;
import com.swak.reactivex.threads.EventLoop;
import com.swak.timer.RTO;
import com.swak.utils.Longs;

import lombok.Getter;

/**
 * PaxosProposer.
 * 
 * @author DELL
 */
@Getter
public class PaxosProposer implements Proposer {

	private final PaxosNode node;
	private final PaxosGroup group;
	private final PaxosConfig config;
	private final PaxosInstance instance;
	private final EventLoop eventLoop;
	private final PaxosStorage storage;
	private final TimeStat timeStat = new TimeStat();

	/**
	 * 每次投票最新的统计信息
	 */
	private final ProposerState state;
	private final BallotStatistics ballotStatistics;

	/**
	 * 当前正在进行的步骤
	 */
	private boolean isPreparing;
	private boolean isAccepting;
	private boolean canSkipPrepare;
	private boolean wasRejectBySomeone;

	/**
	 * 记录 Preparing 或 Accepting 的超时任务.
	 * 如果 当前的 InstanceID 和  timeoutInstanceID 不一致则不需要在创建 timeoutFuture
	 */
	private ScheduledFuture<?> timeoutFuture;
	private long timeoutInstanceID;

	private final Runnable prepareTimeout;
	private final Runnable acceptTimeout;

	/** 超时时间 */
	private long prepareTimeoutMS;
	private long acceptTimeoutMS;

	/** 超时计算规则 */
	private final RTO prepareRTO = new RTO();
	private final RTO acceptRTO = new RTO();

	/** 提升编号的次数 */
	private int proposeCount = 0;

	public PaxosProposer(PaxosConfig config, PaxosNode node, PaxosGroup group, PaxosInstance instance,
			EventLoop eventLoop, PaxosStorage storage) {
		this.node = node;
		this.group = group;
		this.config = config;
		this.instance = instance;
		this.eventLoop = eventLoop;
		this.storage = storage;
		this.state = new ProposerState();
		this.ballotStatistics = new BallotStatistics(group.getSystemSM());
		this.prepareTimeout = this::onPrepareTimeout;
		this.acceptTimeout = this::onAcceptTimeout;
		this.prepareTimeoutMS = this.config.getPrepareTimeoutMs();
		this.acceptTimeoutMS = this.config.getAcceptTimeoutMs();
	}

	@Override
	public void newInstance() {
		this.state.init();
		this.proposeCount = 0;
		this.ballotStatistics.newRound();
		this.cancelTimeout();
	}

	@Override
	public void commit(Value value) {

		ProposerBP.newProposal(value, this.group.getGroupID(), this.instance.getInstanceID());

		/*
		 * 保存提交的数据
		 */
		if (this.state.getValue().isNull()) {
			this.state.setValue(value);
		}

		/*
		 * 如果一直没有被打断过，则可以直接跳入 accept 阶段.
		 */
		if (this.canSkipPrepare && !this.wasRejectBySomeone) {
			ProposerBP.newProposalSkipPrepare(this.group.getGroupID(), this.instance.getInstanceID());
			this.accept();
			return;
		}

		/*
		 * 进入 prepare 阶段.
		 */
		this.prepare(this.wasRejectBySomeone);
	}

	@Override
	public void prepare(boolean needNewBallot) {

		/*
		 * 启动计时器
		 */
		this.timeStat.point();

		/*
		 * 重置一些标志位
		 */
		this.exitAccept();
		this.isPreparing = true;
		this.canSkipPrepare = false;
		this.wasRejectBySomeone = false;

		/*
		 * 重新记录 MaxVote
		 */
		this.state.resetHighestOtherPreAcceptBallot();

		/*
		 * needNewBallot ：来至于 wasRejectBySomeone 默认是 false
		 * newPrepare ：ProposalID +1（先取一个最大值）
		 * ProposalID ：默认是 1
		 */
		if (needNewBallot) {
			this.state.newPrepare();
		}

		/*
		 * 打印 prepare 信息
		 */
		ProposerBP.prepare(this.group.getGroupID(), this.instance.getInstanceID(), this.state.getProposalID());

		/*
		 * 创建 Paxos 消息.
		 */
		PaxosMessage message = new PaxosMessage();
		message.setNodeID(this.node.id());
		message.setType(ProposalType.Prepare);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(this.instance.getInstanceID());
		message.setProposalID(this.state.getProposalID());
		message.setTimestamp(System.currentTimeMillis());

		/*
		 * 开启新一轮统计
		 */
		this.ballotStatistics.newRound();

		/*
		 * 添加第一阶段的超时处理器
		 */
		this.addPrepareTimeout();

		/*
		 * 发送信息
		 */
		this.broadcastMessage(message);
	}

	void addPrepareTimeout() {

		this.cancelTimeout();

		// 记录当前的 timeout 的数据
		this.timeoutFuture = this.eventLoop.schedule(prepareTimeout, this.prepareTimeoutMS, this.config.getTimeUnit());
		this.timeoutInstanceID = this.instance.getInstanceID();
	}

	void onPrepareTimeout() {

		// 不需要继续
		if (this.instance.getInstanceID() != timeoutInstanceID) {
			return;
		}

		ProposerBP.prepareTimeout(this.prepareTimeoutMS, this.group.getGroupID(), this.instance.getInstanceID());
		prepare(this.wasRejectBySomeone);
	}

	private void exitAccept() {
		this.isAccepting = false;
		this.cancelTimeout();
	}

	private void cancelTimeout() {
		if (this.timeoutFuture != null && !this.timeoutFuture.isCancelled() && !this.timeoutFuture.isDone()) {
			this.timeoutFuture.cancel(false);
		}
		this.timeoutFuture = null;
		this.timeoutInstanceID = 0;
	}

	@Override
	public void onPrepareReply(PaxosMessage message) {

		ProposerBP.onPrepareReply(this.group.getGroupID(), this.instance.getInstanceID(), message.getProposalID());

		if (!this.isPreparing) {
			ProposerBP.onPrepareReplyButNotPreparing(this.group.getGroupID(), this.instance.getInstanceID());
			return;
		}

		if (this.state.getProposalID() != message.getProposalID()) {
			ProposerBP.onPrepareReplyNotSameProposalIDMsg(this.group.getGroupID(), this.instance.getInstanceID());
			return;
		}

		this.ballotStatistics.addReceive(message.getNodeID());

		// 没有被拒绝
		if (message.getRejectByPromiseID() == 0) {
			BallotNumber ballot = new BallotNumber(message.getPreAcceptID(), message.getPreAcceptNodeID());
			this.ballotStatistics.addPromiseOrAccept(message.getNodeID());
			this.state.addPreAcceptedValue(ballot, message.getValue());
		}

		// 被拒绝
		else {
			this.ballotStatistics.addReject(message.getNodeID());
			this.wasRejectBySomeone = true;
			this.state.setRejectedProposalID(message.getRejectByPromiseID());
		}

		// 过半通过
		if (this.ballotStatistics.isPassedOnThisRound()) {
			int useTimeMs = this.timeStat.point();
			ProposerBP.preparePass(useTimeMs, this.group.getGroupID(), this.instance.getInstanceID(),
					message.getProposalID());
			this.canSkipPrepare = true;
			this.calculatePrepareRTO(useTimeMs);
			this.cancelTimeout();
			this.accept();
		}

		// 过半拒绝或者都回复了
		// 需要提升编号来继续处理，这时候可以静默一段时间
		else if (this.ballotStatistics.isAllReceiveOnThisRound() || this.ballotStatistics.isRejectedOnThisRound()) {
			int useTimeMs = this.timeStat.point();
			ProposerBP.prepareNotPass(useTimeMs, this.group.getGroupID(), this.instance.getInstanceID(),
					message.getProposalID());
			this.calculatePrepareRTO(useTimeMs);
			this.cancelTimeout();
			this.rePrepare();
		}
	}

	private void calculatePrepareRTO(long useTimeMs) {
		this.prepareRTO.calculate(useTimeMs);
		this.prepareTimeoutMS = prepareRTO.getTimeoutInerval();
		this.prepareTimeoutMS = Longs.bound(this.config.getPrepareTimeoutMs(), this.prepareTimeoutMS,
				this.config.getMaxPrepareTimeoutMs());
	}

	@Override
	public void accept() {

		ProposerBP.accept(this.group.getGroupID(), this.instance.getInstanceID(), this.state.getProposalID());

		/*
		 * 开始计时
		 */
		this.timeStat.point();

		/*
		 * 进入接受阶段
		 */
		exitPrepare();
		this.isAccepting = true;

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setNodeID(this.node.id());
		message.setType(ProposalType.Accept);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(this.instance.getInstanceID());
		message.setProposalID(this.state.getProposalID());
		message.setValue(this.state.getValue());
		message.setTimestamp(System.currentTimeMillis());

		/*
		 * 开启新的一轮
		 */
		this.ballotStatistics.newRound();

		/*
		 * 添加超时
		 */
		this.addAcceptTimeout();

		/*
		 * 发送信息
		 */
		this.broadcastMessage(message);
	}

	private void exitPrepare() {
		this.isPreparing = false;
		this.cancelTimeout();
	}

	private void addAcceptTimeout() {
		this.cancelTimeout();

		// 记录当前的 timeout 的数据
		this.timeoutFuture = this.eventLoop.schedule(acceptTimeout, this.acceptTimeoutMS, this.config.getTimeUnit());
		this.timeoutInstanceID = this.instance.getInstanceID();
	}

	void onAcceptTimeout() {
		/*
		 * 不需要继续
		 */
		if (this.instance.getInstanceID() != timeoutInstanceID) {
			return;
		}

		ProposerBP.acceptTimeout(this.group.getGroupID(), this.instance.getInstanceID());

		/*
		 * 如果一直没有被打断过，则可以直接跳入 accept 阶段.
		 */
		if (this.canSkipPrepare && !this.wasRejectBySomeone) {
			ProposerBP.newProposalSkipPrepare(this.group.getGroupID(), this.instance.getInstanceID());
			this.accept();
			return;
		}

		/*
		 * 否则需要发起第一轮投票
		 */
		prepare(this.wasRejectBySomeone);
	}

	@Override
	public void onAcceptReply(PaxosMessage message) {
		ProposerBP.onAcceptReply(this.group.getGroupID(), this.instance.getInstanceID());

		if (!this.isAccepting) {
			ProposerBP.onAcceptReplyButNotAccepting(this.group.getGroupID(), this.instance.getInstanceID());
			return;
		}

		if (this.state.getProposalID() != message.getProposalID()) {
			ProposerBP.onAcceptReplyNotSameProposalIDMsg(this.group.getGroupID(), this.instance.getInstanceID());
			return;
		}

		this.ballotStatistics.addReceive(message.getNodeID());

		// 被接收（不会返回 MaxVote）
		if (message.getRejectByPromiseID() == 0) {
			this.ballotStatistics.addPromiseOrAccept(message.getNodeID());
		}

		// 被拒绝
		else {
			this.ballotStatistics.addReject(message.getNodeID());
			this.wasRejectBySomeone = true;
			this.state.setRejectedProposalID(message.getRejectByPromiseID());
		}

		// 接收过半处理
		if (this.ballotStatistics.isPassedOnThisRound()) {
			int useTimeMs = this.timeStat.point();
			ProposerBP.acceptPass(useTimeMs, this.group.getGroupID(), this.instance.getInstanceID());
			this.exitAccept();
			this.calculateAcceptRTO(useTimeMs);
			this.proposed();
		}

		// 过半拒绝或者都回复了
		// 需要提升编号来继续处理，这时候可以静默一段时间
		else if (this.ballotStatistics.isRejectedOnThisRound() || this.ballotStatistics.isAllReceiveOnThisRound()) {
			int useTimeMs = this.timeStat.point();
			ProposerBP.acceptNotPass(useTimeMs, this.group.getGroupID(), this.instance.getInstanceID());
			this.exitAccept();
			this.calculateAcceptRTO(useTimeMs);
			this.rePrepare();
		}
	}

	private void calculateAcceptRTO(long useTimeMs) {
		this.acceptRTO.calculate(useTimeMs);
		this.acceptTimeoutMS = acceptRTO.getTimeoutInerval();
		this.acceptTimeoutMS = Longs.bound(this.config.getAcceptTimeoutMs(), this.acceptTimeoutMS,
				this.config.getMaxAcceptTimeoutMs());
	}

	/**
	 * 提升编号，且重新提交
	 */
	private void rePrepare() {

		/*
		 * 记录提升编号的次数, 如果超过最大的次数则静默一段一个提交周期
		 */
		this.proposeCount++;
		if (this.proposeCount >= this.config.getMaxProposeCount()) {
			this.proposeCount = 0;
			this.addPrepareTimeout();
			return;
		}

		/*
		 * 立即提升编号，提交
		 */
		this.prepare(this.wasRejectBySomeone);
	}

	/**
	 * 已经选择的了一个值；可以将此值广播出去，告诉其他的节点 instance 的 值已经确定
	 * 
	 * 对于此 Proposer 来说，可以结束任务，可以去提交下一个值了；
	 * 对于所有的 Proposer 来说，已经确定了一个值，所以不管后续怎么提升编号都只是完成自身的任务而已；
	 */
	@Override
	public void proposed() {

		ProposerBP.proposed(this.group.getGroupID(), this.instance.getInstanceID());

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setNodeID(this.node.id());
		message.setType(ProposalType.Proposed);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(this.instance.getInstanceID());
		message.setProposalID(this.state.getProposalID());
		message.setTimestamp(System.currentTimeMillis());

		/*
		 * 发送消息
		 */
		this.broadcastMessage(message);
	}

	/**
	 * 广播消息：保持顺序性
	 * 
	 * @param message
	 */
	void broadcastMessage(PaxosMessage message) {

		// 通过网络发送消息： 默认不会发送给自己
		this.node.broadcastMessage(message);

		// 当前节点快速的处理消息
		this.instance.doInnerReceivePaxosMessage(message);
	}
}
